/*
	从kvdb中dump出来的文件格式描述如下:
		由数据库表每行记录信息组成
		每行信息对一个字符串，不分行，信息间的分割符号是ascii码 8;
		每行记录信息对应的字符串组成如下：信息id, 信息键名, 信息键值, 键值插入时间文本, 键值插入时间戳, 补齐位
		行信息内部各字段值间的分割字符串为ascii码 0

		信息id相同的多行记录描述了一个完整的对象	


	fullbuild doc文件格式
		对象以<doc>开始，</doc>结束，开始和结束标示符号都单独占1行(\n)
		对象的每一个键值对占用1行，键名=键值
		每一行的末尾都有ascii码 1 结束(即 \01\n )
		对象的第一个键值对是id=idvalue

	计算从程序开始，到fullbuilddoc写入结果文件的时间
*/

#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <math.h>
#include <sys/time.h>

#include <pthread.h>
#include <unistd.h>

int numThread = 0;
//const char* srcFilePath = "/home/jiening/10.252.42.117-t1019_2";
//const char* srcFilePath = "/media/DATA/forCuda/kvdbdata/10.252.42.117-t1019_2";
const char* srcFilePath = "/media/DATA/forCuda/kvdbdata/10.252.42.117-t1030_2";
const char* disFilePath = "/home/jiening/testres.xml";

int sigleProcThread()
{
	numThread = sysconf(_SC_NPROCESSORS_CONF);
	timeval clkBegin;
	gettimeofday(&clkBegin, NULL);
	
	FILE* fp = fopen(srcFilePath, "r");
	FILE* wfp = fopen(disFilePath, "w");
	if(NULL == fp)
	{
		printf("fail to read source kvdb dump file, program over\n");
		return 0;
	}
	if(NULL == wfp)
	{
		printf("fail to open write file\n");
		return 0;
	}
	
	timeval clkFileOpen;
	gettimeofday(&clkFileOpen, NULL);

	char ch;
	int loop=0;
	
	int kvLineCount = 0;		// 已经读取的kvdump的信息行数
	int kvValueCount = 0;		// 当前正在处理的信息字段号（在每个信息行内的）
	int docCount = 0;
	int charmap[256] = {0};
	char buffer[1024*1024];		// buffer
	int bufferpos = 0;
	char nowinfoidstr[64]={0};
	char tmpinfoidstr[64]={0};
	
	while((ch = fgetc(fp)) != EOF)
	{
		if(0 == ch)
		{
			// 信息id
			if(0 == kvValueCount)	
			{
				memcpy(tmpinfoidstr, buffer, bufferpos);
				tmpinfoidstr[bufferpos] = '\0';
				if(0 != strcmp(nowinfoidstr, tmpinfoidstr))
				{
					//新的doc起始
					strcpy(nowinfoidstr, tmpinfoidstr);
					++ docCount;
					if(1 < docCount)
					{
						fprintf(wfp, "</doc>\n");
					}
					//if(10 < docCount){break;}
					fprintf(wfp, "<doc>\nid=%s\n", nowinfoidstr);
				}
			}
			else if(1 == kvValueCount)
			{
				// 键名
				fwrite(buffer, 1, bufferpos, wfp);
				fprintf(wfp, "=");
			}
			else if(2 == kvValueCount)
			{
				fwrite(buffer, 1, bufferpos, wfp);
				fprintf(wfp, "\n");
				// 键值
			}
			else if(3 == kvValueCount)
			{
				// 更新时间文本
			}
			else if(4 == kvValueCount)
			{
				// 更新时间戳
			}

			bufferpos=0;
			++ kvValueCount;
		}
		else if(8 == ch)
		{
			++ kvLineCount;
			kvValueCount = 0;
		}
		else
		{
			buffer[bufferpos] = ch;
			++bufferpos;
		}
		++loop;

		//charmap[ch] += 1;
		//printf("%d %d -> %c\n", loop, ch,ch);
		//if(10000 < loop){break;}
	}
	fprintf(wfp, "</doc>\n");

	fclose(fp);
	fclose(wfp);

	timeval clkEnd;
	gettimeofday(&clkEnd, NULL);
	
	// 打印时间
	long timecostusec = (clkEnd.tv_usec - clkBegin.tv_usec);
	timecostusec += 1000000L*(clkEnd.tv_sec - clkBegin.tv_sec);
	long timecostms = timecostusec/1000L;

	long timecostfopen = (clkFileOpen.tv_usec - clkBegin.tv_usec);	
	timecostfopen += 1000000L*(clkFileOpen.tv_sec - clkBegin.tv_sec);	

	printf("timecostfopen: %ld us\n", timecostfopen);

	printf("\ncompute time: %ld us\n", timecostusec);
	printf("compute time: %ld ms\n", timecostms);

	printf("\nloop: %d \n", loop);
	printf("kvLineCount: %d \n", kvLineCount);
	printf("kvValueCount: %d \n", kvValueCount);
	printf("docCount: %d \n", docCount);
	
	for(int i=0; i<256; ++i)
	{
		break;
		if(0 < charmap[i])
		{
			printf("charmap[%d]=%d\n", i, charmap[i]);
		}
	}


	printf("\ngo and see desinct file %s\n", disFilePath);
	return 0;
}

struct ctinfo
{
	char* rdata;
	long bytenum;
	long offset;
	long cptnum;
	int thdidx;
};


void * fileScanThread(void* pcptinf)
{
	if(NULL == pcptinf)
	{
		return 0;
	}

	struct ctinfo * thectinfo = (struct ctinfo *)pcptinf;

	char disfilePath[128];
	sprintf(disfilePath, "/home/jiening/test.multhd.%d.xml", thectinfo->thdidx);
	//FILE* fp = fopen(srcFilePath, "r");
	//if(NULL == fp)
	//{
	//	printf("in thread %d, fail to read source kvdb dump file, thread over\n", thectinfo->thdidx);
	//	return 0;
	//}
	FILE* wfp = fopen(disfilePath, "w");
	if(NULL == wfp)
	{
		printf("in thread %d, fail to open disfilePath file to write, thread over\n", thectinfo->thdidx);
		return 0;
	}

	printf("in thread %d, file open done, offset=%ld, cptnum=%ld, wfp_addr=%ld \n", thectinfo->thdidx, thectinfo->offset, thectinfo->cptnum, (long)(wfp));
	fflush(stdout);
	//sleep(thectinfo->thdidx);return 0;

	char ch;
	int loop=0;
	
	int kvLineCount = 0;		// 已经读取的kvdump的信息行数
	int kvValueCount = 0;		// 当前正在处理的信息字段号（在每个信息行内的）
	int docCount = 0;
	//int charmap[256] = {0};
	char* buffer = (char*)malloc(1024*64);		// buffer
	int bufferpos = 0;
	char* nowinfoidstr = (char*)malloc(64);
	nowinfoidstr[0] = '\0';
	char* tmpinfoidstr = (char*)malloc(64);
	tmpinfoidstr[0] = '\0';

	printf("thd %d: %d buffer_adr=%ld now_adr=%ld tmp_adr=%ld\n", thectinfo->thdidx, loop, (long)(buffer), (long)(nowinfoidstr), (long)(tmpinfoidstr));
	fflush(stdout);

	//fseek(fp, thectinfo->offset, SEEK_SET);	//文件指针移动到偏移地址
	char* dataread = thectinfo->rdata;
	dataread += thectinfo->offset;
	long readoffset = thectinfo->offset;
	bool finddocend = false;
	bool finddocbegin = false;
	while(true)
	{
		ch = dataread[loop];
		finddocend = false;

		//continue;
		//printf("thd %d: %d %d -> %c bufferpos=%d\n", thectinfo->thdidx, loop, ch, ch, bufferpos);
		//fflush(stdout);
		if(0 == ch)
		{
			// 信息id
			if(0 == kvValueCount)	
			{
				if(bufferpos > 64)
				{
					printf("----bug thd %d: in while %d %d -> %c bufferpos=%d, buffer_adr=%ld before memcpy\n", thectinfo->thdidx, loop, ch, ch, bufferpos, (long)(buffer));fflush(stdout);
				}
				else
				{
					memcpy(tmpinfoidstr, buffer, bufferpos);
					tmpinfoidstr[bufferpos] = '\0';
					if(0 != strcmp(nowinfoidstr, tmpinfoidstr))
					{
						finddocend = true;
						//新的doc起始
						//printf("thd %d: %d %d -> %c bufferpos=%d before strcpy, \n", thectinfo->thdidx, loop, ch, ch, bufferpos);fflush(stdout);
						strcpy(nowinfoidstr, tmpinfoidstr);
						++ docCount;
						if(1 < docCount)
						{
							fprintf(wfp, "</doc>\n");
						}
						if(loop > thectinfo->cptnum)	
						{
							break;
						}
						//if(10 < docCount){break;}
						fprintf(wfp, "<doc>\nid=%s\n", nowinfoidstr);
						finddocbegin = true;
					}
				}
			}
			else if(1 == kvValueCount)
			{
				// 键名
				if(finddocbegin)
				{
					fwrite(buffer, 1, bufferpos, wfp);
					fprintf(wfp, "=");
				}
			}
			else if(2 == kvValueCount)
			{
				// 键值
				if(finddocbegin)
				{
					fwrite(buffer, 1, bufferpos, wfp);
					fprintf(wfp, "\n");
				}
			}
			else if(3 == kvValueCount)
			{
				// 更新时间文本
			}
			else if(4 == kvValueCount)
			{
				// 更新时间戳
			}

			bufferpos=0;
			++ kvValueCount;
		}
		else if(8 == ch)
		{
			++ kvLineCount;
			kvValueCount = 0;
		}
		else
		{
			buffer[bufferpos] = ch;
			++bufferpos;
		}

		++loop;
		++readoffset;

		if(readoffset >= thectinfo->bytenum)
		{
			fprintf(wfp, "</doc>\n");
			break;
		}	
		//charmap[ch] += 1;
		//fflush(stdout);
		//if(10000 < loop){break;}
	}
	free(buffer);
	free(nowinfoidstr);
	free(tmpinfoidstr);

	fclose(wfp);

	printf("\nin thread %d, loop: %d \n",thectinfo->thdidx, loop);
	printf("in thread %d, kvLineCount: %d \n",thectinfo->thdidx, kvLineCount);
	printf("in thread %d, kvValueCount: %d \n",thectinfo->thdidx,kvValueCount);
	printf("in thread %d, docCount: %d \n", thectinfo->thdidx, docCount);
	return 0;
}

int multiThread()
{	
	numThread = sysconf(_SC_NPROCESSORS_CONF);
	timeval clkBegin;
	gettimeofday(&clkBegin, NULL);
	
	FILE* fp = fopen(srcFilePath, "r");
	if(NULL == fp)
	{
		printf("in multiThread, fail to read source kvdb dump file, program over\n");
		return 0;
	}
	fseek(fp, 0L, SEEK_END);
	long fileByteCount = ftell(fp);
	char* freadbuffer = (char*)malloc(fileByteCount);
	fseek(fp, 0L, SEEK_SET);
	size_t readbytecount = fread(freadbuffer, 1, (size_t)fileByteCount, fp);
	if((size_t)fileByteCount != readbytecount)
	{
		printf("in multiThread, fread fail to read %ld bytes but %d byte", fileByteCount, readbytecount);
		return 0;
	}
	fclose(fp);

	timeval clkFileOpen;
	gettimeofday(&clkFileOpen, NULL);

	long bytePerThread = fileByteCount/numThread;

	pthread_t tidp[numThread];
	struct ctinfo cptinf[numThread];
	for(int i=0;i<numThread;++i)
	{
		cptinf[i].rdata = freadbuffer;
		cptinf[i].bytenum = fileByteCount;
		cptinf[i].offset = i*bytePerThread;
		cptinf[i].cptnum = bytePerThread;
		cptinf[i].thdidx = i;
	
		int error = pthread_create(&(tidp[i]),NULL, fileScanThread, (void*)(&(cptinf[i])));
		if(0 != error)
		{
			printf("in multiThread fail to call pthread_create\n");
			return 0;
		}
		else
		{
			printf("in multiThread succ to call pthread_create %d\n", i);
		}
	}

	void* re;
	for(int i=0; i<numThread; ++i)
	{
		printf("in multiThread pthread_join %d\n", i);
		fflush(stdout);
		pthread_join(tidp[i], &re);
	}
	
	free(freadbuffer);
	
	timeval clkEnd;
	gettimeofday(&clkEnd, NULL);

	// 打印时间 
	long timecostusec = (clkEnd.tv_usec - clkBegin.tv_usec);
	timecostusec += 1000000L*(clkEnd.tv_sec - clkBegin.tv_sec);
	long timecostms = timecostusec/1000L;

	long timecostfopen = (clkFileOpen.tv_usec - clkBegin.tv_usec);	
	timecostfopen += 1000000L*(clkFileOpen.tv_sec - clkBegin.tv_sec);	

	printf("\ntimecostfopen: %ld us\n", timecostfopen);

	printf("\ncompute time: %ld us\n", timecostusec);
	printf("compute time: %ld ms\n", timecostms);

	printf("fileByteCount = %ld\n", fileByteCount);
	return 0;
}

int main()
{
	//sigleProcThread();
	multiThread();
}
